package org.quxiao.hbase.test.singleAttrIndex;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

public class SingleAttrIndexManager {

	private static Configuration conf = null;
	private static HBaseAdmin admin = null;
	private static final String IDX_CF_NAME = "srcRow";
	private static final String IDX_QUALIFIER_NAME = "key";
	
	static
	{
		conf = HBaseConfiguration.create();
		try
		{
			admin = new HBaseAdmin(conf);
		}
		catch (ZooKeeperConnectionException zooEx)
		{
			System.err.println("can not connect zookeeper!");
			System.exit(-1);
		}
		catch (MasterNotRunningException masterNotRunEx)
		{
			System.err.println("hbase master isn't running!");
			System.exit(-1);
		}
	}
	
	public static void addIndex (String srcTableName, String cfName, String qualifierName, String idxTableName)
	{
		try
		{
			HTable idxTable = createIndexTable(idxTableName);
			HTable srcTable = new HTable(conf, srcTableName);
			srcTable.setAutoFlush(false);
			Scan srcScan = new Scan();
			srcScan.addColumn(cfName.getBytes(), qualifierName.getBytes());
			ResultScanner srcRs = srcTable.getScanner(srcScan);
			for (Result r: srcRs)
			{
//				System.out.println("read row : " + new String(r.getRow()));
				KeyValue kv = r.getColumnLatest(cfName.getBytes(), qualifierName.getBytes());
				//construct idx row: 
				String idxRowKey = new String(kv.getValue()) + "_" + new String(kv.getRow());
				Put idxPutRow = new Put(idxRowKey.getBytes());
				idxPutRow.add(IDX_CF_NAME.getBytes(), IDX_QUALIFIER_NAME.getBytes(), kv.getRow());
				idxTable.put(idxPutRow);
//				System.out.println("import idx: " + new String(idxPutRow.getRow()));
			}
		}
		catch (IOException ioe)
		{
			ioe.printStackTrace();
			System.exit(-1);
		}
	}
	
	private static HTable createIndexTable (String idxTableName)
	{
		/*
		 * idx table structure:
		 * ------------------
		 * Row	|	CF		|
		 * 		|	srcRow	|
		 * ------------------
		 */
		HTableDescriptor tableDescriptor = new HTableDescriptor(idxTableName.getBytes());
		HColumnDescriptor srcRowFamily = new HColumnDescriptor(IDX_CF_NAME);
		tableDescriptor.addFamily(srcRowFamily);
		HTable idxTable = null;
		try
		{
			if ( admin.tableExists(idxTableName) )
			{
				admin.disableTable(idxTableName);
				admin.deleteTable(idxTableName);
			}
			admin.createTable(tableDescriptor);
			idxTable = new HTable(idxTableName);
		}
		catch (IOException ioe)
		{
			ioe.printStackTrace();
		}
		return idxTable;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub	
		String[] srcDbNames = {/*"Asset_100000", "Asset_500000", "Asset_1000000", "Asset_2000000",*/ "Asset_5000000"};
		String[] idxDbNames = {/*"vendorIdx_Asset_10w", "vendorIdx_Asset_50w", "vendorIdx_Asset_100w", "vendorIdx_Asset_200w", */"vendorIdx_Asset_500w"};
		
		for (int i = 0; i < srcDbNames.length; i ++)
		{
			System.out.println("creating index for " + srcDbNames[i]);
			addIndex(srcDbNames[i], "MetaData", "vendor", idxDbNames[i]);
		}
	}

}
